#include "iomanger.h"

#include <sys/epoll.h>
#include <unistd.h>
#include <fcntl.h>
#include <errno.h>
#include <cstring>

#include "macro.h"


namespace sylar
{

static sylar::Logger::LoggerPtr g_logger = SYLAR_LOG_NAME("system");

IOManger *IOManger::GetThis()
{
    return dynamic_cast<IOManger*>(Scheduler::GetThis());
}

IOManger::IOManger(size_t threads, bool use_coller, const std::string &name)
    : Scheduler(threads, use_coller, name)
{
    m_epfd = epoll_create(5000);
    SYLAR_ASSERT(m_epfd > 0);

    int iRet = pipe(m_tickFds);
    SYLAR_ASSERT(!iRet);

    epoll_event event;
    memset(&event, 0, sizeof(epoll_event));

    event.events = EPOLLIN | EPOLLET;   // 边沿触发
    event.data.fd = m_tickFds[0];

    iRet = fcntl(m_tickFds[0], F_SETFL, O_NONBLOCK);
    SYLAR_ASSERT(!iRet);

    iRet = epoll_ctl(m_epfd, EPOLL_CTL_ADD, m_tickFds[0], &event);
    SYLAR_ASSERT(!iRet);

    contextResize(32);

    start();

}

IOManger::~IOManger()
{
    stop();
    close(m_epfd);
    close(m_tickFds[0]);
    close(m_tickFds[1]);

    size_t size = m_fdContext.size();
    for (size_t i = 0; i < size; i++)
    {
        if (m_fdContext[i])
        {
            delete m_fdContext[i];
        }
    }

}


int IOManger::addEvent(int fd, Event event, std::function<void(void)> cb)
{
    FdContext* fdContext = nullptr;
    RWMutexType::ReadLock rdlock(m_mutex);
    if (static_cast<int>(m_fdContext.size()) > fd)
    {
        fdContext = m_fdContext[fd];
        rdlock.unlock();
    }
    else
    {
        rdlock.unlock();
        RWMutexType::WriteLock wrlock(m_mutex);
        // fd 突然传的很大 size 有问题
        contextResize(fd * 1.5);
        fdContext = m_fdContext[fd];
    }

    FdContext::MutexType::Lock lock(fdContext->mutex);
    if (fdContext->events & event)
    {
        SYLAR_LOG_ERROR(g_logger) << "addEvent assert fd=" << fd 
        << " events"  << event 
        << "fdContext->event: " << fdContext->events;
        SYLAR_ASSERT((fdContext->events & event));
        return -1;
    }
    
    int op = fdContext->events ? EPOLL_CTL_MOD : EPOLL_CTL_ADD;
    epoll_event epollevent;
    memset(&epollevent, 0, sizeof(epollevent));
    epollevent.events = event | fdContext->events | EPOLLET;
    epollevent.data.ptr = static_cast<void*>(fdContext);
    
    int iRet = epoll_ctl(m_epfd, op, fd, &epollevent);
    if (iRet)
    {
        SYLAR_LOG_ERROR(g_logger) << "epoll_ctl error(" << m_epfd << ", "
            << op << ", " << fd << ", " << epollevent.events << " return: " << iRet
            << "errorno" << errno << "error: " << strerror(errno);
        return -1;
    }
    ++m_pendingEventCount;

    fdContext->events = static_cast<Event>(fdContext->events | event);
    FdContext::EventContext& eventConetxt = fdContext->getEventContext(event);
    SYLAR_ASSERT(!eventConetxt.scheduler && !eventConetxt.fiber && !eventConetxt.cb);

    eventConetxt.scheduler = Scheduler::GetThis();
    if (cb)
    {
        eventConetxt.cb.swap(cb);
    }
    else
    {
        eventConetxt.fiber = Fiber::GetThis();
        SYLAR_ASSERT(eventConetxt.fiber->getState() == Fiber::EXEC);
    }

    return 0;
}

bool IOManger::delEvent(int fd, Event event)
{
    RWMutexType::ReadLock rdlock(m_mutex);

    if ((int)m_fdContext.size() <= fd)
    {
        return false;
    }
    FdContext* fd_ctx = m_fdContext[fd];
    rdlock.unlock();

    FdContext::MutexType lock(fd_ctx->mutex);
    if (!(fd_ctx->events & event))
    {
        return false;
    }

    Event newEvent = static_cast<Event>(fd_ctx->events & ~event);
    int op = newEvent ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;
    
    epoll_event epollEvent;
    memset(&epollEvent, 0 ,sizeof(epoll_event));
    epollEvent.events = EPOLLET | newEvent;
    epollEvent.data.ptr = fd_ctx;

    int iRet = epoll_ctl(m_epfd, op, fd, &epollEvent);
    if (iRet)
    {
        SYLAR_LOG_ERROR(g_logger) << "epoll_ctl error(" << m_epfd << ", "
            << op << ", " << fd << ", " << epollEvent.events << " return: " << iRet
            << "errorno" << errno << "error: " << strerror(errno);
        return false;
    }
    --m_pendingEventCount;
    fd_ctx->events = newEvent;

    FdContext::EventContext& eventContext = fd_ctx->getEventContext(event);
    fd_ctx->resetContext(eventContext);
    return true;
}

bool IOManger::cancelEvent(int fd, Event event)
{
    RWMutexType::ReadLock rdlock(m_mutex);

    if ((int)m_fdContext.size() <= fd)
    {
        return false;
    }
    FdContext* fd_ctx = m_fdContext[fd];
    rdlock.unlock();

    FdContext::MutexType lock(fd_ctx->mutex);
    if (!(fd_ctx->events & event))
    {
        return false;
    }

    Event newEvent = static_cast<Event>(fd_ctx->events & ~event);
    int op = newEvent ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;
    
    epoll_event epollEvent;
    memset(&epollEvent, 0 ,sizeof(epoll_event));
    epollEvent.events = EPOLLET | newEvent;
    epollEvent.data.ptr = fd_ctx;

    int iRet = epoll_ctl(m_epfd, op, fd, &epollEvent);
    if (iRet)
    {
        SYLAR_LOG_ERROR(g_logger) << "epoll_ctl error(" << m_epfd << ", "
            << op << ", " << fd << ", " << epollEvent.events << " return: " << iRet
            << "errorno" << errno << "error: " << strerror(errno);
        return false;
    }

    fd_ctx->triggerEvent(event);
    --m_pendingEventCount;
    return true;
}

bool IOManger::cancelAll(int fd)
{
    RWMutexType::ReadLock rdlock(m_mutex);

    if ((int)m_fdContext.size() <= fd)
    {
        return false;
    }
    FdContext* fd_ctx = m_fdContext[fd];
    rdlock.unlock();

    FdContext::MutexType lock(fd_ctx->mutex);
    if (!fd_ctx->events)
    {
        return false;
    }

    int op = EPOLL_CTL_DEL;
    
    epoll_event epollEvent;
    memset(&epollEvent, 0 ,sizeof(epoll_event));
    epollEvent.events = 0;
    epollEvent.data.ptr = fd_ctx;

    int iRet = epoll_ctl(m_epfd, op, fd, &epollEvent);
    if (iRet)
    {
        SYLAR_LOG_ERROR(g_logger) << "epoll_ctl error(" << m_epfd << ", "
            << op << ", " << fd << ", " << epollEvent.events << " return: " << iRet
            << "errorno" << errno << "error: " << strerror(errno);
        return false;
    }

    if (fd_ctx->events & READ)
    {
        fd_ctx->triggerEvent(READ);
        --m_pendingEventCount;
    }

    if (fd_ctx->events & WRITE)
    {
        fd_ctx->triggerEvent(WRITE);
        --m_pendingEventCount;
    }

    SYLAR_ASSERT(fd_ctx->events == 0);
    return true;
}

void IOManger::tickle()
{
    if (!hasidleThreads())
    {
        return ;
    }
    int rt = write(m_tickFds[1], "T", 1);
    SYLAR_ASSERT(rt == 1);
}

bool IOManger::stopping()
{
    uint64_t timeout = 0;
    return stopping(timeout);
}

bool IOManger::stopping(uint64_t &timeout)
{
    timeout = getNextTimer();
    return timeout == ~0ull && m_pendingEventCount == 0 && Scheduler::stopping();
}

void IOManger::idle()
{
    epoll_event* events = new epoll_event[64];

    std::shared_ptr<epoll_event> shard_events(events, [](epoll_event* events){
        delete[] events;
    }) ;

    while(true)
    {
        uint64_t next_timeout = 0;
        if(stopping(next_timeout))
        {
            SYLAR_LOG_INFO(g_logger) << "name: " << getName()
                                                     << "idel is stop exit";
            break;
        }
        
        int iRet =  0;
        do
        {
            static const int MAX_TIMEOUT = 3000;
            
            if (next_timeout != ~0ull)
            {
                //next_timeout = (int)next_timeout > MAX_TIMEOUT ? MAX_TIMEOUT : next_timeout;
                next_timeout = (int)next_timeout > MAX_TIMEOUT
                                ? MAX_TIMEOUT : next_timeout;
            }
            else
            {
                next_timeout = MAX_TIMEOUT;
            }

            iRet = epoll_wait(m_epfd, events, 64, (int)next_timeout);
            if (iRet < 0 && errno == EINTR)
            {

            }
            else
            {
               break;
            }
        } while (true);
        
        std::vector<std::function<void()> > cbs;
        listExpiredCb(cbs);
        if(!cbs.empty()) 
        {
            //SYLAR_LOG_DEBUG(g_logger) << "on timer cbs.size=" << cbs.size();
            schedule(cbs.begin(), cbs.end());
            cbs.clear();
        }

        for (int i = 0; i < iRet; i++)
        {
            epoll_event& event = events[i];
            if (event.data.fd == m_tickFds[0])
            {
                uint8_t dummy = 0;
                while (1 == read(m_tickFds[0], &dummy, sizeof(uint8_t)))
                {
                   
                }
                continue;
            }
            FdContext* fd_ctx = static_cast<FdContext*>(event.data.ptr);

            FdContext::MutexType::Lock lock(fd_ctx->mutex);
            if (event.events & (EPOLLERR | EPOLLHUP)) 
            {   // 错误或中断
                event.events |= (EPOLLIN | EPOLLOUT) & fd_ctx->events;
            }
            int real_event  = NONE;
            if (event.events & EPOLLIN)
            {
                real_event |= READ;
            }
            
            if(event.events & EPOLLOUT)
            {
                real_event |= WRITE;
            }

            if ((fd_ctx->events & real_event) == NONE)
            {
                continue;
            }

            int other_events = (fd_ctx->events & ~real_event);
            int op = other_events ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;
            event.events = EPOLLET | other_events;

            int iRet2 = epoll_ctl(m_epfd, op, fd_ctx->fd, &event);
            if (iRet2)
            {
                SYLAR_LOG_ERROR(g_logger) << "epoll_ctl error(" << m_epfd << ", "
                    << op << ", " << fd_ctx->fd << ", " << event.events << " return: " << iRet2
                    << "errorno" << errno << "error: " << strerror(errno);
                continue;
            }

            if (real_event & READ)
            {
                fd_ctx->triggerEvent(READ);
                --m_pendingEventCount;
            }

            if (real_event & WRITE)
            {
                fd_ctx->triggerEvent(WRITE);
                --m_pendingEventCount;
            }
        }
                    
        Fiber::ptr cur = Fiber::GetThis();
        auto raw_cur = cur.get();
        cur.reset();
        //SYLAR_LOG_ERROR(g_logger) << " getFiberNumber " << getFiberNumber();
        raw_cur->swapOut();
    }

}
void IOManger::contextResize(size_t size)
{
    m_fdContext.resize(size);

    for (size_t i  = 0; i < m_fdContext.size(); ++i)
    {
        if (!m_fdContext[i])
        {
            m_fdContext[i] = new FdContext;
            m_fdContext[i]->fd = i;
        }
    }
}

void IOManger::onTimerInserFront()
{
    tickle();
}

IOManger::FdContext::EventContext &IOManger::FdContext::getEventContext(Event event)
{
    switch (event)
    {
    case IOManger::Event::READ:
        return read;
        break;
    case IOManger::Event::WRITE:
        return write;
    default:
        SYLAR_ASSERT2(false, "getEvent Context event:" +  std::to_string(static_cast<int>(event)));
        break;
    }
}


void IOManger::FdContext::resetContext(EventContext &context)
{
    context.scheduler = nullptr;
    context.fiber.reset();
    context.cb = nullptr;
}


void IOManger::FdContext::triggerEvent(Event event)
{
    SYLAR_ASSERT(events & event);

    events = (Event)(events & ~event);

    EventContext& event_ctx = getEventContext(event);

    // 已经将事件回调加入调度队列 后续都应该清空才对吧

    if (event_ctx.scheduler)
    {
        if (event_ctx.cb)
        {
            event_ctx.scheduler->schedule(event_ctx.cb);
        }
        else
        {
            event_ctx.scheduler->schedule(event_ctx.fiber);
            
        }

        event_ctx.scheduler = nullptr;
        // 这里应该也要至空
        event_ctx.fiber = nullptr;
        event_ctx.cb = nullptr;
    }
}


}